package com.kgc.bigdata.spark.sql

import org.apache.spark.{SparkConf, SparkContext}

/**
  * DataFrame和RDD互操作
  */
object DataFrameRDDApp {

  def main(args: Array[String]) {
    //method1()

    method2()
  }

  /**
    * 第二种操作方式：通过编程接口指定Schema
    */
  def method2(): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]")
      .setAppName("DataFrameRDDApp")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    import sqlContext.implicits._
    val people = sc.textFile("H:/workspace/SparkProject/src/data/people.txt")

    // 以字符串的方式定义DataFrame的Schema信息
    val schemaString = "name age"

    //导入所需要的类
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.types.{StructType, StructField, StringType}

    // 根据自定义的字符串schema信息产生DataFrame的Schema
    val schema =
      StructType(
        schemaString.split(" ").map(fieldName =>
          StructField(fieldName, StringType, true)))

    //将RDD转换成Row
    val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))

    // 将Schema作用到RDD上
    val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)

    // 将DataFrame注册成临时表
    peopleDataFrame.registerTempTable("people")

    /**
      * 获取name字段的值
      *
      * 运行结果为：
      * Name: Michael
      * Name: Andy
      * Name: Justin
      *
      * 其他的常用操作和第一种方式一样
      */
    val results = sqlContext.sql("SELECT name FROM people")
    results.map(t => "Name: " + t(0)).collect().foreach(println)

    sc.stop()
  }

  /**
    * 第一种操作方式：使用反射获取RDD内的Schema
    */
  def method1(): Unit = {

    val sparkConf = new SparkConf().setMaster("local[2]")
      .setAppName("DataFrameRDDApp")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    import sqlContext.implicits._

    //将RDD转成DataFrame
    val people = sc.textFile("H:/workspace/SparkProject/src/data/people.txt")
      .map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()

    /**
      * 使用DataFrame API访问
      *
      * 运行结果为：
      * +-------+---+
      * |   name|age|
      * +-------+---+
      * |Michael| 29|
      * |   Andy| 30|
      * | Justin| 19|
      * +-------+---+
      */
    people.show()

    /**
      * 将DataFrame注册成临时表，后续可以直接使用SQL进行查询
      *
      * 运行结果为：满足年龄条件的只有一条记录
      * +------+---+
      * |  name|age|
      * +------+---+
      * |Justin| 19|
      * +------+---+
      */
    people.registerTempTable("people")
    val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")
    teenagers.show()


    /**
      * DataFrame转成RDD进行操作：根据索引号取值
      * 运行结果为：Name: Justin
      */

    teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

    /**
      * DataFrame转成RDD进行操作：根据字段名称取值
      * 运行结果为：Name: Justin
      */
    teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)

    /**
      * DataFrame转成RDD进行操作：一次返回多列的值
      *
      * 运行结果为： Map(name -> Justin, age -> 19)
      */
    teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)


    sc.stop()
  }

  /**
    * 定义Person类
    *
    * @param name 姓名
    * @param age  年龄
    */
  case class Person(name: String, age: Int)

}
